package logcollector

import (
	"github.com/hpcloud/tail"
	"context"
	"github.com/golang/glog"
	"cloud_worm/worm/client/monitor/kafka"
	"sync"
	"github.com/astaxie/beego/logs"
	"bytes"
	"cloud_worm/worm/client/monitor/monitorlog"
)

type LogReader struct {
	FileName 		string
	Parser	 		string

	tail 			*tail.Tail
	producer 		kafka.KafProducer
	logFd 			*logs.BeeLogger

	cancel 			context.CancelFunc
}

func (l *LogReader) OpenTail(cancel context.CancelFunc, producer kafka.KafProducer, logFd *logs.BeeLogger) error {
	t, err := tail.TailFile(l.FileName, tail.Config{
		ReOpen: true,
		Follow: true,
		MustExist: false,
		Poll: true,
		Location: &tail.SeekInfo{Offset: 0, Whence: 2},
	})
	if err != nil {
		return err
	}

	l.logFd = logFd
	l.producer = producer
	l.tail = t
	l.cancel = cancel
	return nil
}

func (l *LogReader) Stop()  {
	if l.cancel != nil {
		l.cancel()				// 触发资源回收
	}
	return
}

func (l *LogReader) Loop(ctx context.Context, wg *sync.WaitGroup) {
	go func(l *LogReader) {
		glog.Infof("Doing start read '%s' log", l.FileName)
		wg.Add(1)

		var sep = []byte("")
		var msgFlag = []byte("afocus_syslog")
		sl := make([][]byte, 0, 2)
		sl = append(sl, msgFlag)
		
		for {
			select {
			case msg, ok := <- l.tail.Lines:
				if l.logFd != nil || l.producer != nil {
					if ok {
						// 一行日志的处理
						var logInfo LogInfo
						if err := ExecLogParser(l.Parser, []byte(msg.Text), &logInfo); err != nil {
							if err == noNeedAnalyzerErr {
								// 不做处理
							} else {
								glog.Errorf("'%s' parser error: %s", l.Parser, err.Error())
							}
						} else {
							if b, err := logInfo.MarshalJSON(); err == nil {
								if l.logFd != nil {
									// 写日志
									l.logFd.Info("%s: %s", monitorlog.MonitorLogCollector, b)
								}
								if l.producer != nil {
									// 写kafka
									sl = sl[:1]
									sl = append(sl, b)
									l.producer.AsyncSend(bytes.Join(sl, sep), "")
								}
							}
						}
					}
				} else {
					// 没有消息出处，则停止分析
					glog.Infof("stop read '%s' log, for no output", l.FileName)
					wg.Done()
					if l.tail != nil {
						l.tail.Stop()
					}
					return
				}

			case <- ctx.Done():
				glog.Infof("Done stop read '%s' log", l.FileName)
				wg.Done()
				if l.tail != nil {
					l.tail.Stop()
				}
				return
			}
		}
	}(l)

	return
}